package com.abel.bigwater.kmgw

import com.abel.bigwater.data.BcdString
import com.abel.bigwater.dataserver.model.FlowDataPojo
import com.abel.bigwater.dataserver.model.PacketNotEnoughException
import com.abel.bigwater.dataserver.model.PojoHelper
import com.abel.bigwater.dataserver.model.ServiceConfig
import com.abel.bigwater.dataserver.model.kming.*
import com.abel.bigwater.kmgw.dao.ConfigDao
import com.abel.bigwater.kmgw.dao.FlowDao
import io.netty.buffer.Unpooled
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component
import java.net.InetSocketAddress
import java.net.StandardSocketOptions
import java.nio.ByteBuffer
import java.nio.channels.AsynchronousChannelGroup
import java.nio.channels.AsynchronousSocketChannel
import java.util.*
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantReadWriteLock
import kotlin.concurrent.read
import kotlin.concurrent.write

@Component
class KmingVirtualDeviceBean {
    @Autowired
    var configDao: ConfigDao? = null

    @Autowired
    var iwgDao: FlowDao? = null

    var cfg: ServiceConfig? = null

    var curThread: Thread? = null

    /**
     * lock for service
     */
    val lock = ReentrantReadWriteLock()

    var _running: Boolean = true
    var running: Boolean
        get() {
            lock.read { return _running }
        }
        set(value) {
            lock.write { _running = value }
        }

    /**
     * start in a thread.
     */
    fun startService() {
        running = true

        curThread = Thread {
            // 等待kming-server启动完成
            Thread.sleep(10 * 1000)

            startSend()
        }.also {
            it.isDaemon = true
            it.start()
            it.join(100)
        }

        lgr.info("$KMING_GW_NAME 已启动.")
    }

    /**
     * stop the service.
     */
    fun stopService() {
        running = false

        curThread?.apply {
            interrupt()
            join(200)
        }

        lgr.info("$KMING_GW_NAME 已停止.")
    }

    /**
     * 发送主函数
     */
    fun startSend() {
        while (!Thread.currentThread().isInterrupted && running) {
            var socket: AsynchronousSocketChannel? = null
            var cur: FlowDataPojo? = null

            try {
                cfg = configDao!!.getConfig(KMING_CLIENT_CONFIG_ID)!!
                cfg!!.running = true

                val unsentList = iwgDao?.unsentList()
                lgr.info("待转发消息: ${unsentList?.size}")

                for (fd in unsentList.orEmpty()) {
                    cur = fd

                    // convert packages
                    val plst = PojoHelper.toKmingPacket(fd, null)

                    if (plst.isEmpty()) {
                        lgr.warn("解析消息为空: ${fd.station} @ ${fd.startTime}")
                        continue
                    }

                    // try to connect
                    val tp = Executors.newFixedThreadPool(1)
                    val agroup = AsynchronousChannelGroup.withThreadPool(tp)

                    socket = AsynchronousSocketChannel.open(agroup)
                    socket.setOption(StandardSocketOptions.TCP_NODELAY, true)

                    val fcon = socket.connect(InetSocketAddress(cfg!!.remoteHost, cfg!!.remotePort!!.toInt()))

                    lgr.info("消息发送中: ${fd.station} @ ${fd.startTime} ...")

                    fcon[10, TimeUnit.SECONDS]

                    if (!fcon.isDone || fcon.isCancelled || socket?.isOpen == false) {
                        lgr.error("连接超时: ${fd.station} @ ${fd.startTime}")

                        iwgDao?.updateSendCount(fd)
                        continue
                    }

                    val rst1 = sendReqPacket(socket, plst[0])
                    lgr.info("got response from server for send-req: ${rst1}")
                    if (rst1?.data !is KmingResponseData ||
                            (rst1.data as KmingResponseData).sessionReponseType != SessionReponseType.Success) {
                        lgr.error("发送请求包失败: ${fd.station} @ ${fd.startTime}")

                        iwgDao?.updateSendCount(fd)
                        return
                    }

                    val rst2 = sendDataPacket(socket, plst[1])
                    lgr.info("got response from server for send-data:$rst2")
                    if (rst2?.data !is KmingResponseData ||
                            (rst2.data as KmingResponseData).sessionReponseType != SessionReponseType.Success) {
                        lgr.error("发送数据包失败: ${fd.station} @ ${fd.startTime}")

                        iwgDao?.updateSendCount(fd)
                        return
                    } else {
                        lgr.info("发送数据成功: ${fd.station} @ ${fd.startTime}")
                        iwgDao?.markSent(fd)
                    }

                    // send shutdown
                    val shut = sendShutdownReq(socket, fd, plst[1])
                    lgr.info("got response from server for shutdown: $shut")
                    if (shut?.data is KmingRepondShutdown) {
                        if ((shut.data as KmingRepondShutdown).sessionReponseType == SessionReponseType.Success) {
                            lgr.info("主站允许关闭连接: ${fd.station} @ ${fd.startTime}")

                            socket!!.shutdownOutput()
                            Thread.sleep(100)
                            socket.shutdownInput()
                            socket.close()
                        } else {
                            // wait for command.
                        }
                    }

                    Thread.sleep(10)
                }

                Thread.sleep(1000L * cfg!!.sleepSeconds)
            } catch (ex: Exception) {
                lgr.error("${KMING_GW_NAME} 暂时故障: ${ex.message}, 稍后重试...", ex)
                Thread.sleep(1000L * Random().nextInt(32))

                if (cur != null) {
                    iwgDao?.updateSendCount(cur)
                }
            } finally {
                try {
                    socket?.close()

                    Thread.sleep(10)
                } catch (ex: Exception) {
                    // nothing.
                }
            }
        }
    }

    private fun sendShutdownReq(socket: AsynchronousSocketChannel, fd: FlowDataPojo, snd: KmingPacket): KmingClientPacket? {
        val kp = KmingPacket().apply {
            termCode = snd.termCode
            sessionStatus.serialNo = (snd.sessionStatus.serialNo ?: 0) + 1

            data = TermShutdownData()
            dataLen = data!!.bufLength()
        }

        val buf = Unpooled.buffer()
        kp.pack(buf)

        ByteBuffer.wrap(ByteArray(buf.readableBytes()).also {
            buf.readBytes(it)
        }).also {
            val f = socket.write(it)
            f[2, TimeUnit.SECONDS]
        }

        val bb2 = ByteBuffer.allocate(2048)
        var cntSeconds = 0
        while (cntSeconds < 10) {
            cntSeconds += 2
            val f2 = socket.read(bb2)
            f2[2, TimeUnit.SECONDS]

            val bb3 = bb2.duplicate().apply { flip() }
            buf.clear().writeBytes(ByteArray(bb3.remaining()).also { bb3.get(it) })
            lgr.info("got response from server for shutdown: ${BcdString.hexString(buf, withBlank = true)}")

            try {
                val rply = KmingClientPacket()
                if (rply.parseBuf(buf)) return rply
            } catch (ex: PacketNotEnoughException) {
                // wait for more.
            }

            Thread.sleep(1000)
        }

        return null
    }

    /**
     * send data and return response packet.
     */
    private fun sendDataPacket(socket: AsynchronousSocketChannel, first: KmingPacket): KmingClientPacket? {
        val buf = Unpooled.buffer()
        first.pack(buf)
        val bb = ByteBuffer.wrap(ByteArray(buf.readableBytes()).apply { buf.readBytes(this) })
        val fw1 = socket.write(bb)
        fw1[10, TimeUnit.SECONDS]

        if (!fw1.isDone || fw1.isCancelled) {
            lgr.error("发送消息1超时: ${first.termCode.fmt}")
            socket.close()
            Thread.sleep(1000)
            return null
        }

        val bb2 = ByteBuffer.allocate(2048)

        var totalSeconds = 0
        val rst1 = KmingClientPacket()
        while (totalSeconds < 10) {
            var ftr = socket.read(bb2)
            ftr[2, TimeUnit.SECONDS]
            totalSeconds += 2

            val ob = bb2.duplicate().flip() as ByteBuffer
            if (ob.hasRemaining()) {
                val buf2 = Unpooled.buffer()
                ByteArray(ob.remaining()).also {
                    ob.get(it)
                    buf2.writeBytes(it)
                }

                try {
                    if (rst1.parseBuf(buf2)) {
                        bb2.clear().flip()
                        break
                    }
                } catch (ex: PacketNotEnoughException) {
                    lgr.error("wait for next bytes: ${ex.message}")
                }
            }
        }

        // parse good
        return rst1
    }

    /**
     * send request packet, and return response packet.
     */
    fun sendReqPacket(socket: AsynchronousSocketChannel, first: KmingPacket): KmingClientPacket? {
        val buf = Unpooled.buffer()
        first.pack(buf)
        val bb = ByteBuffer.wrap(ByteArray(buf.readableBytes()).apply { buf.readBytes(this) })
        val fw1 = socket.write(bb)
        fw1[10, TimeUnit.SECONDS]

        if (!fw1.isDone || fw1.isCancelled) {
            lgr.error("发送消息1超时: ${first.termCode.fmt}")
            socket.close()
            Thread.sleep(1000)
            return null
        }

        val bb2 = ByteBuffer.allocate(2048)

        var totalSeconds = 0
        val rst1 = KmingClientPacket()
        while (totalSeconds < 10) {
            var ftr = socket.read(bb2)
            ftr[2, TimeUnit.SECONDS]

            totalSeconds += 2

            val ob = bb2.duplicate().flip() as ByteBuffer

            if (ob.hasRemaining()) {
                val buf2 = Unpooled.buffer()
                ByteArray(ob.remaining()).also {
                    ob.get(it)
                    buf2.writeBytes(it)
                }

                lgr.info("got bytes from kming-server: ${BcdString.hexString(buf2, withBlank = true)}")
                try {
                    if (rst1.parseBuf(buf2)) {
                        bb2.clear().flip()
                        break
                    }
                } catch (ex: PacketNotEnoughException) {
                    lgr.error("wait for next bytes: ${ex.message}")
                }
            }
        }

        // parse good
        return rst1
    }

    companion object {
        const val KMING_CLIENT_CONFIG_ID = "kmcgeClient"
        const val KMING_GW_NAME = "昆明通用大表网关"

        private val lgr = LoggerFactory.getLogger(KmingVirtualDeviceBean::class.java)
    }
}
